package com.atguigu.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author yhm
 * @create 2024-03-30 11:03
 */
public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        // TODO1 构建Flink的执行环境  流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // TODO2 读取数据
        DataStreamSource<String> lineDs = env.readTextFile("input/words.txt");

        // TODO3 拆分单词
        SingleOutputStreamOperator<Tuple2<String, Long>> flatMapOp = lineDs.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(new Tuple2<>(word, 1L));
                }
            }
        });

        // TODO4 聚合单词
        SingleOutputStreamOperator<Tuple2<String, Long>> reduceDs = flatMapOp
//                .keyBy(tuple -> tuple.f0)
                .keyBy(new KeySelector<Tuple2<String, Long>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Long> value) throws Exception {
                        return value.f0;
                    }
                })
//                .reduce(new ReduceFunction<Tuple2<String, Long>>() {
//                    @Override
//                    public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
//                        return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
//                    }
//                });
        .sum(1);

        reduceDs.print();

        // 流处理  必须要有一个开始执行的指令
        env.execute();


    }
}
